package com.shujia.dws

import com.shujia.common.utils.SparkTool
import com.shujia.common.utils.poly.Polygon
import org.apache.spark.sql.expressions.{UserDefinedFunction, Window}
import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}

object DwsScenicTouristMskDay extends SparkTool {
  override def run(spark: SparkSession): Unit = {
    import spark.implicits._
    import org.apache.spark.sql.functions._


    val isInScenicBoundaryWithGrid: UserDefinedFunction = udf((grid_id: Long, boundary: String) => {
      new Polygon(boundary).contains(grid_id)
    })

    // 加载位置数据融合表
    val mergeLocDF: DataFrame = spark.table("dwi.dwi_res_regn_mergelocation_msk_d")
      .where($"day_id" === day_id)

    // 加载景区网格配置表
    val scenicGridDF: DataFrame = spark.table("dim.dim_scenic_grid")
      // 便于通过网格id进行关联，需要进行行转列
      .select($"scenic_id", $"scenic_name", $"grids", explode(split($"grids", ",")) as "grid_id")


    // 加载用户画像表
    val userTagDF: DataFrame = spark.table("dim.dim_usertag_msk_d").where($"day_id" === day_id)


    // 基于位置数据融合表及景区网格配置表将属于景区内的位置记录过滤出来
    mergeLocDF
      .join(scenicGridDF, "grid_id") // 能通过网格id关联上说明该位置在景区内
      // 计算停留时间
      // 计算用户进入某景区的最早的时间 用户离开某景区的最晚的时间
      .groupBy($"mdn", $"scenic_id", $"scenic_name", $"grids")
      .agg(min($"start_date") as "min_start_date", max($"end_date") as "max_end_date")
      // 统计每个用户在每个景区的停留时间
      .withColumn("d_stay_time", diffTime($"max_end_date", $"min_start_date"))
      .where($"d_stay_time" > 30 * 60)
      // 关联用户画像数据 获取用户来源地
      .join(userTagDF, "mdn")
      // 过滤出居住地不在景区内地游客
      .where(!array_contains(split($"grids",","),$"resi_grid_id" ))
      .select($"mdn"
        , $"resi_county_id" as "source_county_id"
        , $"scenic_id" as "d_scenic_id"
        , $"scenic_name" as "d_scenic_name"
        , $"min_start_date" as "d_arrive_time"
        , round($"d_stay_time" / 60 / 60, 2) as "d_stay_time"
      )
      .distinct()
      .write
      .format("csv")
      .option("sep", ",")
      .mode(SaveMode.Overwrite)
      .save(s"/daas/motl/dws/dws_scenic_tourist_msk_d/day_id=$day_id")

    // 添加分区
    spark.sql(
      s"""
         |alter table dws.dws_scenic_tourist_msk_d  add if not exists partition(day_id='$day_id')
         |""".stripMargin)


  }
}
